package com.zzk.www.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zzk.www.domain.Message;
import com.zzk.www.domain.MessageVo;
import com.zzk.www.domain.User;
import com.zzk.www.domain.UserVo;
import com.zzk.www.nettyServer.tool.ChannelUserHolder;
import com.zzk.www.service.MessageHelperService;
import com.zzk.www.service.MessageDaoService;
import com.zzk.www.service.MsgSendService;
import com.zzk.www.tool.QueueService;
import com.zzk.www.tool.cache.CacheService;
import com.zzk.www.tool.common.BeanHolder;
import com.zzk.www.tool.common.DateUtil;
import com.zzk.www.tool.common.SystemUtil;
import com.zzk.www.tool.constants.CacheConstant;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Service
public class MessageHelperServiceImpl implements MessageHelperService {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    private static final Logger LOGGER = Logger.getLogger(MessageHelperServiceImpl.class);

    @Autowired
    private CacheService cacheService;

    @Autowired
    private QueueService queueService;

	@Autowired
	private MessageDaoService messageDaoService;

	@Autowired
	private MsgSendService msgSendService;

    @Value("${msg.batchStoreSize}")
    private Integer batchStoreSize;

    private String batchNo = SystemUtil.getIp() + "_" + DateUtil.getFmt(new Date().getTime(), "yyyy-MM-dd HH:mm");
    /**
     * 是否可以处理消息
     */
    private boolean canProcessMsg = true;

    private boolean finishProcessMsg = false;


    @Override
    public void backgroudTaskProcess() {
        ExecutorService ec = Executors.newFixedThreadPool(1);

        // 发送消息存储处理
        ec.execute(new Runnable() {
            @Override
            public void run() {
                doStoreMessage();
            }
        });


        //
        ec.shutdown();

        finishProcessMsg = true;
    }



    //验证用户(在握手成功后执行回调接口代码)
    @Override
    public void validate(ChannelHandlerContext ctx, FullHttpRequest request) {
        String uri = request.getUri();
        String token = "";
        int index = uri.indexOf("?token=");
        if (index == -1 || StringUtils.isEmpty(uri.substring(index))) {
            ctx.channel().writeAndFlush(new TextWebSocketFrame("验证失败【token不能为空】,将以匿名账号登录"));
//            ctx.channel().close();
            ChannelUserHolder.put(ctx.channel(), new UserVo());
            return;
        }
        token = uri.substring(index);
        String tokenKey = CacheConstant.Auth.token + "_" + token;
        UserVo userVo = (UserVo) BeanHolder.get(CacheService.class, "redisCacheService").get(tokenKey);
        if (userVo == null || !(userVo instanceof UserVo)) {
            ctx.channel().writeAndFlush(new TextWebSocketFrame("验证失败【token不合法】,将以匿名账号登录"));
//            ctx.channel().close();
            ChannelUserHolder.put(ctx.channel(), new UserVo());
            return;
        }
        //如果已经登陆,关闭上个连接
        Channel connectedChannel = ChannelUserHolder.getChannelByUserId(userVo.getUserId());
        if (connectedChannel != null ) {
            ChannelUserHolder.remove(connectedChannel);
            connectedChannel.writeAndFlush(new TextWebSocketFrame("你的账号已在别的地点登陆,请重新登陆"));
            connectedChannel.close();
        }
        //登录成功
        ChannelUserHolder.put(ctx.channel(), userVo);

        handleOffLineMsg(ctx,userVo);

        handleOffLineFriend(ctx,userVo);

    }

    //处理离线消息
    private void handleOffLineMsg(ChannelHandlerContext ctx, UserVo userVo) {
        String offLineListKey = CacheConstant.Msg.userOffLineMsgList + "_" + userVo.getUserId();
        String offLineMsg;
        while (ctx.channel().isOpen() && StringUtils.isNotEmpty(offLineMsg = queueService.pop(offLineListKey))) {
            //发送消息
            offLineMsg = msgSendService.sendMsg(ctx.channel(), userVo, offLineMsg, true);

            //发送成功，将消息放到已发送队列
            String sentMsgListKey = CacheConstant.Msg.sentMsgList + "_" + userVo.getUserId();
            queueService.add(sentMsgListKey, offLineMsg);
        }

    }

    //处理离线好友请求
    private void handleOffLineFriend(ChannelHandlerContext ctx, UserVo userVo) {
        String friendOffLineListKey = CacheConstant.Msg.friendSyncOffLineMsgList + "_" + userVo.getUserId();
        String offLineMsg;
        while (ctx.channel().isOpen() && StringUtils.isNotEmpty(offLineMsg = queueService.pop(friendOffLineListKey))) {
            //发送消息
            msgSendService.sendFriendSyncMsg(ctx.channel(),userVo,offLineMsg,true);
        }
    }


    //接收到消息
    @Override
    public void receiveMessage(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        Channel channel = ctx.channel();
        Message message;
        try {
            message = MAPPER.readValue(msg.text(), Message.class);

        } catch (IOException e) {
            channel.writeAndFlush(new TextWebSocketFrame("消息格式不正确"));
            e.printStackTrace();
            throw new RuntimeException("============消息格式不正确============"+msg.text());
        }

    }

    //用户连接(在验证后,可用可不用)
    @Override
    public void userConnected(ChannelHandlerContext ctx) {
        //获得客户端通道
        Channel incoming = ctx.channel();

        User user = ChannelUserHolder.get(ctx.channel());

        String userName = user == null ? "匿名用戶(" + incoming.remoteAddress() + ")" : user.getUserName();

        for (Channel channel : ChannelUserHolder.keys()) {
            if (channel != incoming) {
                channel.writeAndFlush("欢迎:[" + userName + "]进入聊天室\n");
            }
        }
    }

    //用户离开
    @Override
    public void userDisconnected(ChannelHandlerContext ctx) {
        ChannelUserHolder.remove(ctx.channel());
    }


    public String sendMsg(YTWebSocket ytWebSocket, UserVo userVo, String message, boolean online) {
        MessageVo msgBean = JsonUtils.json2Object(message, MessageVo.class);
        if (userVo != null) {
            msgBean.setReceiveNickname(userVo.getNickname());
            msgBean.setReceiveIp(userVo.getClientIp());
            msgBean.setReceiveServerIp(userVo.getServerIp());
        }
        if (online) {
            ResponseObject responseObject = ResponseObjectUtils.buildResObject();
            responseObject.setBizType("" + MsgBizType.msg.getCode());
            msgBean.setReceiveTime(new Date().getTime());
            responseObject.setData(msgBean);
            String json = JsonUtils.object2Json(responseObject);
            if (ytWebSocket.getWebSocket().getReadyState().name().equals(WebSocket.READYSTATE.OPEN.name())) {
                ytWebSocket.getWebSocket().send(json);
            }
        }
        String batchNo = msgBean.getMsgId();
        String typeText = online ? "在线" : "离线";
        String title = msgBean.getReceiveNickname() + " 已接收到【" + typeText + "】消息";
        HbLogContextMgr.writeSuccesMsg("im", "msg", batchNo, title);
        return JsonUtils.object2Json(msgBean);
    }



    /**
     * 客户端发送消息，无需处理状态
     *
     * @param conn
     * @param message
     */
    @Override
    public void onMessage(WebSocket conn, String message) {
        String id = conn.getRemoteSocketAddress().toString();
        ZZKWebSocket zzkWebSocket = sysWebSocketMap.get(id);
        if (!canProcessMsg) {
            writeSystemMsg(getServerBatchNo(), "服务器升级处理,拒绝消息发送", zzkWebSocket.toString());
            conn.close(1, "服务器升级处理,拒绝消息发送"); // 关闭
            return;
        }
        String key = zzkWebSocket.getKey();
        if (zzkWebSocket.getStatus().intValue() == 0) {// 已下线
            String errMsg = "connection is not validate，message=" + message + ",key=" + key;
            msgSendService.sendFeedbackErrorMsg(zzkWebSocket, "当前连接已下线", AppConstant.ResponseCode.CONN_OFFLINE, errMsg);
            return;
        }
        if (zzkWebSocket.getAuthStatus().intValue() == 0) {//
            String errMsg = "connection is not validate，message=" + message + ",key=" + key;
            msgSendService.sendFeedbackErrorMsg(zzkWebSocket, "发送消息没有权限", SystemConstant.ResponseCode.NOT_AUTHORITY,
                    errMsg);
            return;
        }

        MessageVo msgBean = JsonUtils.json2Object(message, MessageVo.class);

        // 获取发送用户信息
        String sendUserKey = CacheConstant.Auth.loginUser + "_" + msgBean.getSendUserId();
        UserVo sendUserVo = (UserVo) cacheService.get(sendUserKey);

        msgBean.setSendIp(sendUserVo.getClientIp());
        msgBean.setSendServerIp(SystemUtil.getIp());
        msgBean.setSendNickname(sendUserVo.getNickname());
        msgBean.setSendTime(new Date().getTime());
        msgBean.setSendUserAvatar(sendUserVo.getAvatar());
        msgBean.setSendUserMobile(sendUserVo.getAccountNo());

        // 获取接收用户信息
        List<Long> receiveUserIds = msgBean.getReceiveUserIds();
        if (msgBean.getReceiveUserId() != null) {
            receiveUserIds.add(msgBean.getReceiveUserId());
        }

        // 发送多个用户处理
        String listKey;
        for (Long receiveUserId : receiveUserIds) {
            String userKey = CacheConstant.Auth.loginUser + "_" + receiveUserId;
            UserVo userVo = (UserVo) cacheService.get(userKey);
            msgBean.setReceiveUserId(receiveUserId);

            if (userVo != null) {
                listKey = CacheConstant.Msg.readyMsgList + "_" + userVo.getServerIp();
            } else {
                listKey = CacheConstant.Msg.userOffLineMsgList + "_" + receiveUserId;
            }
            msgBean.setStatus(MessageStatus.ready.getCode());

            String msg = JsonUtils.object2Json(msgBean);
            queueService.add(listKey, msg);

            msgSendService.doSendMsg(zzkWebSocket, msgBean, message);

        }

    }


    /**
     * 服务端发送好友通讯录同步消息
     */
    protected void doSendFriendMessage() {
        String listKey = CacheConstant.Msg.friendSyncMsgList + "_" + SystemUtil.getIp();
        while (canProcessMsg) {
            try {
                final String message = queueService.pop(listKey);
                if (message == null) {
                    Thread.sleep(1000);
                    continue;
                }
                friendSyncSendTask(message);
            } catch (Exception e) {
                writeSystemErrorMsg("doSendFriendMessage", "发送好友同步消息任务处理出错", ExceptionUtils.getPrintStackTrace(e));
            }
        }

    }

    private void friendSyncSendTask(String message) {
        FriendVo friendVo = JsonUtils.json2Object(message, FriendVo.class);
        Long receiveUserId = friendVo.getUserId();
        ZZKWebSocket zzkWebSocket = userWebSocketMap.get(receiveUserId);

        boolean msgSendStatus = false;
        if (zzkWebSocket != null) {
            if (zzkWebSocket.getWebSocket().getReadyState().name().equals(WebSocket.READYSTATE.OPEN.name())) {
                msgSendService.sendFriendSyncMsg(zzkWebSocket, null, message, true);
                msgSendStatus = true;
            }
        }

        if (!msgSendStatus) {
            String sentMsgListKey = CacheConstant.Msg.friendSyncOffLineMsgList + "_" + SystemUtil.getIp();
            queueService.add(sentMsgListKey, message);

            msgSendService.sendFriendSyncMsg(zzkWebSocket, null, message, false);
        }
    }

    //服务端关闭前一个连接
    protected void doClosePrevConn() {
        String listKey = CacheConstant.Msg.closePrevConnMsgList + "_" + SystemUtil.getIp();
        while (canProcessMsg) {
            try {
                final String message = queueService.pop(listKey);
                if (message == null) {
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                    }
                    continue;
                }
                Long userId = Long.valueOf(message);
                ZZKWebSocket zzkWebSocket = userWebSocketMap.get(userId);
                if (zzkWebSocket != null) {
                    zzkWebSocket.setStatus(0);
                    msgSendService.sendFeedbackErrorMsg(zzkWebSocket, "关闭【异点】服务器的前一个连接", AppConstant.ResponseCode.CONN_OFFLINE, zzkWebSocket.toString());
                }

            } catch (Exception e) {
            }
        }
    }

    protected void doStoreMessage() {
        String listKey = CacheConstant.Msg.sentMsgList + "_" + SystemUtil.getIp();
        List<Message> dataList = new ArrayList<>(350);
        while (canProcessMsg) {
            try {
                final String message = queueService.pop(listKey);
                if (message == null) {
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                    }
                    continue;
                }
                MessageVo msgBeanVo = JsonUtils.json2Object(message, MessageVo.class);
                Message msgBean = BeanUtils.beanCopy(msgBeanVo, Message.class);
                msgBean.setStatus(MessageStatus.sent.getCode());
                dataList.add(msgBean);
                if (dataList.size() >= batchStoreSize) {
                    int size = dataList.size();
                    messageDaoService.insertBatch(dataList);
                    writeSystemMsg(getServerBatchNo(), "存储已发送数据成功,size=" + size, "");
                    dataList.clear();
                }
            } catch (Exception e) {
                writeSystemErrorMsg("doStoreMessage", "数据存储任务处理出错", ExceptionUtils.getPrintStackTrace(e));
            }
        }

        //最后处理尾数

        int size = dataList.size();
        if (size > 0) {
            messageDaoService.insertBatch(dataList);
            writeSystemMsg(getServerBatchNo(), "存储已发送数据成功,已结束,size=" + size, "");
        }

    }

    /**
     * 服务端发送消息
     */
    private void doSendMessage() {
        final long startMainTime = System.currentTimeMillis();
        ExecutorService ec = new ThreadPoolExecutor(4, 4, 0, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()) {
            @Override
            protected void terminated() {
                super.terminated();
                System.out.println("finish, take time = " + (System.currentTimeMillis() - startMainTime));
            }
        };
        String listKey = CacheConstant.Msg.readyMsgList + "_" + SystemUtil.getIp();
        while (canProcessMsg) { //
            // 消息处理开始
            try {
                final String message = queueService.pop(listKey);
                if (message == null) { // 没有消息，歇一会儿
                    try {
                        Thread.sleep(1000);
                    } catch (Exception e) {
                    }
                    continue;
                }
                ec.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            messageSendTask(message);
                        } catch (Exception e) {
                            writeSystemErrorMsg("doSendMessage", "发送单个消息任务处理出错", ExceptionUtils.getPrintStackTrace(e));
                        }
                    }
                });

            } catch (Exception e) {
                writeSystemErrorMsg("doSendMessage", "发送消息任务处理出错", ExceptionUtils.getPrintStackTrace(e));
            }
        }
        // 结束处理
        ec.shutdown();
    }

    private void messageSendTask(String message) {
        MessageVo msgBean = JsonUtils.json2Object(message, MessageVo.class);
        Long receiveUserId = msgBean.getReceiveUserId();
        ZZKWebSocket zzkWebSocket = userWebSocketMap.get(receiveUserId);
        boolean msgSendStatus = false;
        if (zzkWebSocket != null) {
            if (zzkWebSocket.getWebSocket().getReadyState().name().equals(WebSocket.READYSTATE.OPEN.name())) {
                msgSendStatus = true;
            } else { // 连接已关闭
                // TODO
            }
        } else { // 没有找到
            // TODO
        }

        if (msgSendStatus) {

            String userKey = CacheConstant.Auth.loginUser + "_" + receiveUserId;
            UserVo userVo = (UserVo) cacheService.get(userKey);
            message = msgSendService.sendMsg(zzkWebSocket, userVo, message, true);

            //根据本机ip创建本机发送消息队列
            String sentMsgListKey = CacheConstant.Msg.sentMsgList + "_" + SystemUtil.getIp();
            queueService.add(sentMsgListKey, message);

        } else {
            //根据接收用户id创建离线消息队列
            String offLineListKey = CacheConstant.Msg.userOffLineMsgList + "_" + msgBean.getReceiveUserId();
            queueService.add(offLineListKey, message);
        }
    }

    @Override
    public void destory() {
        writeSystemErrorMsg(getServerBatchNo(), "服务器升级处理,资源释放", "");
        canProcessMsg = false; // 停止处理消息

        // 客户端发给服务端，不必要关心消息是否成功
        // 服务端必须保证发送完成
        // 客户端发送不成功，服务端要通知客户端

    }

    @Override
    public boolean isDestory() {
        return finishProcessMsg;
    }

    @Override
    public Map<String, ZZKWebSocket> getSysWebSocketMap() {
        return sysWebSocketMap;
    }

    @Override
    public String getServerBatchNo() {
        return batchNo;
    }

    @Override
    public void writeSystemErrorMsg(String batchNo, String message, String errMsg) {
    }

    private void writeSystemMsg(String serverBatchNo, String string, String string2) {

    }

}
